package io.kiki.sba.registry.server.metadata.cleaner;


import io.kiki.sba.registry.cache.ConsecutiveSuccess;
import io.kiki.sba.registry.common.model.console.PersistenceData;
import io.kiki.sba.registry.common.model.console.PersistenceDataBuilder;
import io.kiki.sba.registry.common.model.constants.ValueConstants;
import io.kiki.sba.registry.common.model.metaserver.cleaner.AppRevisionSlice;
import io.kiki.sba.registry.common.model.metaserver.cleaner.AppRevisionSliceRequest;
import io.kiki.sba.registry.common.model.store.AppRevision;
import io.kiki.sba.registry.server.metadata.MetadataLeaderService;
import io.kiki.sba.registry.server.metadata.bootstrap.config.ServerConfig;
import io.kiki.sba.registry.server.metadata.provide.data.ProvideDataService;
import io.kiki.sba.registry.server.metadata.remoting.session.DefaultSessionServerService;
import io.kiki.sba.registry.store.api.DBResponse;
import io.kiki.sba.registry.store.api.OperationStatus;
import io.kiki.sba.registry.store.api.date.DateNowRepository;
import io.kiki.sba.registry.store.api.repository.AppRevisionRepository;
import io.kiki.sba.registry.store.jdbc.config.MetadataConfig;
import io.kiki.sba.registry.store.jdbc.convertor.AppRevisionDomainConvertor;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.JsonUtils;
import io.kiki.sba.registry.util.StringFormatter;
import io.kiki.sba.registry.util.WakeUpLoopExecuteTask;
import io.kiki.stack.netty.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.scheduling.annotation.Scheduled;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.TimeUnit;

public class AppRevisionCleaner implements MetadataLeaderService.LeaderElectorListener, ApplicationListener<ContextRefreshedEvent> {
    private static final Logger logger = LoggerFactory.getLogger(AppRevisionCleaner.class);

    private static final boolean DEFAULT_ENABLED = true;
    final Renewer renewer = new Renewer();
    final Cleaner cleaner = new Cleaner();
    private final int slotNum = 256;
    @Autowired
    DateNowRepository dateNowRepository;
    @Autowired
    AppRevisionRepository appRevisionRepository;
    @Autowired
    MetadataConfig metadataConfig;
    @Autowired
    DefaultSessionServerService sessionServerService;
    @Autowired
    MetadataLeaderService metadataLeaderService;
    @Autowired
    ProvideDataService provideDataService;
    @Autowired
    ServerConfig serverConfig;
    ConsecutiveSuccess consecutiveSuccess;
    private int lastSlotId = -1;

    public AppRevisionCleaner() {
    }

    @PostConstruct
    public void init() {
        consecutiveSuccess = new ConsecutiveSuccess(slotNum * 3, (long) metadataConfig.getRevisionRenewIntervalMinutes() * 60 * 1000 * 5);
        metadataLeaderService.registerListener(this);
    }

    public void start() {
        ConcurrentUtils.createDaemonThread(AppRevisionCleaner.class.getSimpleName() + "-renewer", renewer).start();
        ConcurrentUtils.createDaemonThread(AppRevisionCleaner.class.getSimpleName() + "-cleaner", cleaner).start();
        logger.info("AppRevisionCleaner started");
    }

    void renew() {
        if (!metadataLeaderService.amILeader()) {
            return;
        }
        int slotId = nextSlotId();
        Collection<AppRevisionSlice> list = new ArrayList<>();
        try {
            for (Object result : sessionServerService.broadcastInvoke(new AppRevisionSliceRequest(slotNum, slotId), 1000 * 30).values()) {
                list.add((AppRevisionSlice) result);
            }
            for (String revision : AppRevisionSlice.merge(list).getRevisions()) {
                appRevisionRepository.heartbeatDB(revision);
                ConcurrentUtils.sleepUninterruptibly(10, TimeUnit.MILLISECONDS);
            }
            consecutiveSuccess.success();
        } catch (Throwable e) {
            logger.error("renew app revisions failed:", e);
            consecutiveSuccess.fail();
        }
    }

    Date dateBeforeNow(int minutes) {
        Calendar calendar = Calendar.getInstance();
        calendar.setTime(dateNowRepository.getNow());
        calendar.add(Calendar.MINUTE, -minutes);
        return calendar.getTime();
    }

    void markDeleted() {
        if (!metadataLeaderService.amILeader()) {
            return;
        }
        if (!consecutiveSuccess.check()) {
            return;
        }
        List<AppRevision> expired = appRevisionRepository.getExpired(dateBeforeNow(metadataConfig.getRevisionRenewIntervalMinutes() * 5), serverConfig.getAppRevisionMaxRemove());
        // before markDeleted refresh app revision switch
        appRevisionSwitchRefresh();
        for (AppRevision revision : expired) {
            revision.setDeleted(true);
            try {
                appRevisionRepository.replace(revision);
                logger.info("mark deleted revision: {}", revision.getRevision());
                ConcurrentUtils.sleepUninterruptibly(10, TimeUnit.MILLISECONDS);
            } catch (Throwable e) {
                logger.error("mark deleted revision failed: {}", revision.getRevision(), e);
            }
        }
    }

    private void appRevisionSwitchRefresh() {
        DBResponse<PersistenceData> ret = provideDataService.queryProvideData(ValueConstants.APP_REVISION_WRITE_SWITCH_DATA_ID);
        AppRevisionDomainConvertor.EnableConfig enableConfig = null;
        if (ret.getOperationStatus() == OperationStatus.success) {
            PersistenceData data = ret.getEntity();
            String switchString = data.getData();
            if (StringUtils.isNotBlank(switchString)) {
                try {
                    enableConfig = JsonUtils.read(switchString, AppRevisionDomainConvertor.EnableConfig.class);
                } catch (Throwable e) {
                    logger.error("Decode appRevision write switch failed", e);
                }
            }
        }
        if (enableConfig != null) {
            logger.info("appRevisionSwitch prev={}/{}", AppRevisionDomainConvertor.getEnableConfig().isServiceParams(), AppRevisionDomainConvertor.getEnableConfig().isServiceParamsLarge());
            AppRevisionDomainConvertor.setEnableConfig(enableConfig);
            logger.info("appRevisionSwitch update={}/{}", enableConfig.isServiceParams(), enableConfig.isServiceParamsLarge());
        }
    }

    @Scheduled(initialDelay = 60000, fixedRate = 60000)
    public void digestAppRevision() {
        if (!metadataLeaderService.amILeader()) {
            return;
        }
        try {
            Map<String, Integer> counts = appRevisionRepository.countByApp();
            for (Map.Entry<String, Integer> entry : counts.entrySet()) {
                String app = entry.getKey();
                int count = entry.getValue();
                if (count >= serverConfig.getAppRevisionCountAlarmThreshold()) {
                    logger.info("[AppRevisionCountAlarm]app={},count={}", app, count);
                }
            }
        } catch (Throwable e) {
            logger.error("[AppRevisionCounter] digest failed: ", e);
        }
    }

    void cleanup() {
        if (!metadataLeaderService.amILeader()) {
            return;
        }
        if (!consecutiveSuccess.check()) {
            return;
        }
        int count = appRevisionRepository.cleanDeleted(dateBeforeNow(metadataConfig.getRevisionRenewIntervalMinutes() * 10), serverConfig.getAppRevisionMaxRemove());
        if (count > 0) {
            logger.info("clean up {} revisions", count);
        }
    }

    private boolean isEnabled() {
        DBResponse<PersistenceData> ret = provideDataService.queryProvideData(ValueConstants.APP_REVISION_CLEANER_ENABLED_DATA_ID);
        if (ret.getOperationStatus() == OperationStatus.success) {
            PersistenceData data = ret.getEntity();
            return Boolean.parseBoolean(data.getData());
        }
        return DEFAULT_ENABLED;
    }

    public void setEnabled(boolean enabled) {
        PersistenceData persistenceData = PersistenceDataBuilder.createPersistenceData(ValueConstants.APP_REVISION_CLEANER_ENABLED_DATA_ID, Boolean.toString(enabled));
        try {
            provideDataService.saveProvideData(persistenceData);
        } catch (Exception e) {
            logger.error("set app revision cleaner failed: ", e);
            throw new RuntimeException(StringFormatter.format("set app revision cleaner failed: {}", e.getMessage()));
        }
    }

    synchronized int nextSlotId() {
        lastSlotId = (lastSlotId + 1) % slotNum;
        return lastSlotId;
    }

    @Override
    public void becomeLeader() {
        consecutiveSuccess.clear();
    }

    @Override
    public void loseLeader() {
        consecutiveSuccess.clear();
    }

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        start();
    }

    final class Renewer extends WakeUpLoopExecuteTask {
        @Override
        public int getWaitingMillis() {
            return (metadataConfig.getRevisionRenewIntervalMinutes() * 1000 * 60 / slotNum);
        }

        @Override
        public void _execute_() {
            renew();
        }
    }

    final class Cleaner extends WakeUpLoopExecuteTask {
        @Override
        public int getWaitingMillis() {
            int base = metadataConfig.getRevisionRenewIntervalMinutes() * 1000 * 60;
            return (int) (base + Math.random() * base);
        }

        @Override
        public void _execute_() {
            if (!isEnabled()) {
                return;
            }
            markDeleted();
            cleanup();
        }
    }
}
